package preprocess;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import utils.Utils;

public class SelectBuyer extends Configured implements Tool{
	
	public static String getSelectedBuyerPath(String date) {
		return Utils.rootDir + date + "/SelectBuyer/";
	}
	
	public static class PvSearchClickMap extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// TODO Auto-generated method stub
			String[] values = value.toString().split(Utils.separator);
			k.set(values[3]);
			v.set(value);
			output.collect(k, v);
		}
		
	}
	
	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		private Text v = new Text();

		@Override
		public void reduce(Text key, Iterator<Text> iter,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// TODO Auto-generated method stub
			int pvCount = 0;
			int clickCount = 0;
			while (iter.hasNext()) {
				String str = iter.next().toString();
				++pvCount;
				if (pvCount > 300)
					return;
				String[] values = str.split(Utils.separator);
				String[] clicks = values[29].split("\u0002");
				for (String click : clicks) {
					if (Utils.isValid(click))
						++clickCount;
					if (clickCount > 300)
						return;
				}
			}
			if (pvCount < 5 || clickCount < 5)
				return;
			v.set(Utils.getRecord(new String[]{String.valueOf(pvCount), String.valueOf(clickCount)},
					Utils.separator));
			output.collect(key, v);
		}
		
	}
	@Override
	public int run(String[] args) throws Exception {
		// input
		String pvSearchClickFile = PvSearchClickJoiner.getPvSearchClickDetailPath(args[0]);
		
		// Output
		String output = getSelectedBuyerPath(args[0]);
		System.out.println("start job SelectUser..");
		System.out.println("pvSearchClickFile: " + pvSearchClickFile);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		JobConf conf = new JobConf(getConf(), SelectBuyer.class);
		conf.setJobName("SelectUser");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);

		MultipleInputs.addInputPath(conf, 
				new Path(pvSearchClickFile), 
				TextInputFormat.class, 
				PvSearchClickMap.class);
		
		FileOutputFormat.setOutputPath(conf, new Path(output));
		conf.setReducerClass(Reduce.class);
		JobClient.runJob(conf);
		return 0;
	}
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		int res = ToolRunner.run(new Configuration(), new SelectBuyer(),
				args);
		System.exit(res);
	}

}
